package com.excesys.config;

import com.fasterxml.jackson.core.JsonEncoding;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import org.apache.commons.beanutils.PropertyUtils;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.document.DocumentField;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.data.domain.Pageable;
import org.springframework.data.elasticsearch.ElasticsearchException;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.ScriptedField;
import org.springframework.data.elasticsearch.config.ElasticsearchConfigurationSupport;
import org.springframework.data.elasticsearch.core.*;
import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
import org.springframework.data.elasticsearch.core.aggregation.impl.AggregatedPageImpl;
import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.charset.Charset;
import java.util.*;

/**
 * ES 配置文件
 */
@Configuration
public class ElasticSearchConfig extends ElasticsearchConfigurationSupport {

    @Bean
    public Client elasticsearchClient() throws UnknownHostException {
        Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build();
        TransportClient client = new PreBuiltTransportClient(settings);
        //9300端口：ES节点之间通讯使用
        client.addTransportAddress(new TransportAddress(InetAddress.getByName("localhost"), 9300));
        return client;
    }

    @Bean
    public ElasticsearchTemplate elasticsearchTemplate() throws UnknownHostException {
        return new ElasticsearchTemplate(elasticsearchClient(), new MyResultMapper(entityMapper()));
    }

    // use the ElasticsearchEntityMapper
    @Bean
    @Override
    public EntityMapper entityMapper() {
        ElasticsearchEntityMapper entityMapper = new ElasticsearchEntityMapper(elasticsearchMappingContext(), new DefaultConversionService());
        entityMapper.setConversions(elasticsearchCustomConversions());
        return entityMapper;
    }

    //这里需要自己实现结果映射解析不然无法高亮
    class MyResultMapper extends AbstractResultMapper {

        private final MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext;

        public MyResultMapper() {
            this(new SimpleElasticsearchMappingContext());
        }

        public MyResultMapper(MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext) {

            super(new DefaultEntityMapper(mappingContext));

            Assert.notNull(mappingContext, "MappingContext must not be null!");

            this.mappingContext = mappingContext;
        }

        public MyResultMapper(EntityMapper entityMapper) {
            this(new SimpleElasticsearchMappingContext(), entityMapper);
        }

        public MyResultMapper(
                MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext,
                EntityMapper entityMapper) {

            super(entityMapper);

            Assert.notNull(mappingContext, "MappingContext must not be null!");

            this.mappingContext = mappingContext;
        }

        @Override
        public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {

            long totalHits = response.getHits().getTotalHits();
            float maxScore = response.getHits().getMaxScore();

            List<T> results = new ArrayList<>();
            for (SearchHit hit : response.getHits()) {
                if (hit != null) {
                    T result = null;
                    if (!StringUtils.isEmpty(hit.getSourceAsString())) {
                        result = mapEntity(hit.getSourceAsString(), clazz);
                    } else {
                        result = mapEntity(hit.getFields().values(), clazz);
                    }

                    setPersistentEntityId(result, hit.getId(), clazz);
                    setPersistentEntityVersion(result, hit.getVersion(), clazz);
                    setPersistentEntityScore(result, hit.getScore(), clazz);

                    populateScriptFields(result, hit);

                    results.add(result);
                }
            }

            return new AggregatedPageImpl<T>(results, pageable, totalHits, response.getAggregations(), response.getScrollId(),
                    maxScore);
        }

        private String concat(Text[] texts) {
            StringBuilder sb = new StringBuilder();
            for (Text text : texts) {
                sb.append(text.toString());
            }
            return sb.toString();
        }


        private <T> void populateScriptFields(T result, SearchHit hit) {
            if (hit.getFields() != null && !hit.getFields().isEmpty() && result != null) {
                for (java.lang.reflect.Field field : result.getClass().getDeclaredFields()) {
                    ScriptedField scriptedField = field.getAnnotation(ScriptedField.class);
                    if (scriptedField != null) {
                        String name = scriptedField.name().isEmpty() ? field.getName() : scriptedField.name();
                        DocumentField searchHitField = hit.getFields().get(name);
                        if (searchHitField != null) {
                            field.setAccessible(true);
                            try {
                                field.set(result, searchHitField.getValue());
                            } catch (IllegalArgumentException e) {
                                throw new ElasticsearchException(
                                        "failed to set scripted field: " + name + " with value: " + searchHitField.getValue(), e);
                            } catch (IllegalAccessException e) {
                                throw new ElasticsearchException("failed to access scripted field: " + name, e);
                            }
                        }
                    }
                }
            }

            for (HighlightField field : hit.getHighlightFields().values()) {
                try {
                    PropertyUtils.setProperty(result, field.getName(), concat(field.fragments()));
                } catch (InvocationTargetException | IllegalAccessException | NoSuchMethodException e) {
                    throw new ElasticsearchException("failed to set highlighted value for field: " + field.getName()
                            + " with value: " + Arrays.toString(field.getFragments()), e);
                }
            }
        }

        private <T> T mapEntity(Collection<DocumentField> values, Class<T> clazz) {
            return mapEntity(buildJSONFromFields(values), clazz);
        }

        private String buildJSONFromFields(Collection<DocumentField> values) {
            JsonFactory nodeFactory = new JsonFactory();
            try {
                ByteArrayOutputStream stream = new ByteArrayOutputStream();
                JsonGenerator generator = nodeFactory.createGenerator(stream, JsonEncoding.UTF8);
                generator.writeStartObject();
                for (DocumentField value : values) {
                    if (value.getValues().size() > 1) {
                        generator.writeArrayFieldStart(value.getName());
                        for (Object val : value.getValues()) {
                            generator.writeObject(val);
                        }
                        generator.writeEndArray();
                    } else {
                        generator.writeObjectField(value.getName(), value.getValue());
                    }
                }
                generator.writeEndObject();
                generator.flush();
                return new String(stream.toByteArray(), Charset.forName("UTF-8"));
            } catch (IOException e) {
                return null;
            }
        }

        @Override
        public <T> T mapResult(GetResponse response, Class<T> clazz) {
            T result = mapEntity(response.getSourceAsString(), clazz);
            if (result != null) {
                setPersistentEntityId(result, response.getId(), clazz);
                setPersistentEntityVersion(result, response.getVersion(), clazz);
            }
            return result;
        }

        @Override
        public <T> LinkedList<T> mapResults(MultiGetResponse responses, Class<T> clazz) {
            LinkedList<T> list = new LinkedList<>();
            for (MultiGetItemResponse response : responses.getResponses()) {
                if (!response.isFailed() && response.getResponse().isExists()) {
                    T result = mapEntity(response.getResponse().getSourceAsString(), clazz);
                    setPersistentEntityId(result, response.getResponse().getId(), clazz);
                    setPersistentEntityVersion(result, response.getResponse().getVersion(), clazz);
                    list.add(result);
                }
            }
            return list;
        }

        private <T> void setPersistentEntityId(T result, String id, Class<T> clazz) {

            if (clazz.isAnnotationPresent(Document.class)) {

                ElasticsearchPersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(clazz);
                ElasticsearchPersistentProperty idProperty = persistentEntity.getIdProperty();

                // Only deal with String because ES generated Ids are strings !
                if (idProperty != null && idProperty.getType().isAssignableFrom(String.class)) {
                    persistentEntity.getPropertyAccessor(result).setProperty(idProperty, id);
                }
            }
        }

        private <T> void setPersistentEntityVersion(T result, long version, Class<T> clazz) {

            if (clazz.isAnnotationPresent(Document.class)) {

                ElasticsearchPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(clazz);
                ElasticsearchPersistentProperty versionProperty = persistentEntity.getVersionProperty();

                // Only deal with Long because ES versions are longs !
                if (versionProperty != null && versionProperty.getType().isAssignableFrom(Long.class)) {
                    // check that a version was actually returned in the response, -1 would indicate that
                    // a search didn't request the version ids in the response, which would be an issue
                    Assert.isTrue(version != -1, "Version in response is -1");
                    persistentEntity.getPropertyAccessor(result).setProperty(versionProperty, version);
                }
            }
        }

        private <T> void setPersistentEntityScore(T result, float score, Class<T> clazz) {

            if (clazz.isAnnotationPresent(Document.class)) {

                ElasticsearchPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(clazz);

                if (!entity.hasScoreProperty()) {
                    return;
                }

                entity.getPropertyAccessor(result) //
                        .setProperty(entity.getScoreProperty(), score);
            }
        }
    }

}
